package com.my.study.controller;

import com.my.study.bean.PrintBean;
import com.my.study.utils.JSONUtils;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

import static com.my.study.common.constant.TopicConstants.*;

/**
 * 消息Controller
 *
 * @author: yidujun
 * @create: 2021/07/23 10:16
 */
@RestController
@RequestMapping("/msg")
public class MsgController {

    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    @SuppressWarnings("unchecked")
    public void setKafkaTemplate(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @RequestMapping("/send")
    public void sendMsg() {
        PrintBean printBean = new PrintBean();
        printBean.setMsg("这是一个消息！！！");
        String value = JSONUtils.toJsonString(printBean);
//        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC12, value);

        for (int i = 0; i < 10; i++) {
            kafkaTemplate.send(TOPIC12, value);
        }

//        // 同步发送消息
//        try {
//            SendResult<String, String> sendResult = future.get();
//            ProducerRecord<String, String> producerRecord = sendResult.getProducerRecord();
//            RecordMetadata recordMetadata = sendResult.getRecordMetadata();
//
//            String sendValue = producerRecord.value();
//            Integer partition = producerRecord.partition();
//
//            System.out.println(sendResult.toString());
//        } catch (InterruptedException | ExecutionException e) {
//            e.printStackTrace();
//        }

    }


}
